package com.zhang.spark_1.spark_core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 * @title:
 * @author: zhang
 * @date: 2021/12/8 23:05 
 */
object Spark04_top10_ACC {

  def main(args: Array[String]): Unit = {
    //获取spark的连接
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("top10")
    val sc: SparkContext = new SparkContext(conf)
    //读取数据
    val rdd: RDD[String] = sc.textFile("data/user_visit_action.txt")
    //封装累加器
    val hcACC = new HotCategoryACC
    sc.register(hcACC, "hcACC")

    rdd.foreach(
      line => {
        val datas = line.split("_")
        if (datas(6) != "-1") {
          hcACC.add((datas(6), "click"))
        } else if (datas(8) != "null") {
          val ids: Array[String] = datas(8).split(",")
          ids.foreach(
            id => {
              hcACC.add((id, "order"))
            }
          )
        } else if (datas(10) != "null") {
          val ids: Array[String] = datas(10).split(",")
          ids.foreach(
            id => {
              hcACC.add((id, "pay"))
            }
          )
        }
      }
    )

    val accVal: mutable.Map[String, HotCategory] = hcACC.value

    val categories: mutable.Iterable[HotCategory] = accVal.map(_._2)
    categories.toList.sortWith(
      (left, right) => {
        if (left.clickCnt > right.clickCnt) {
          true
        } else if (left.clickCnt == right.clickCnt) {
          if (left.orderCnt > right.orderCnt) {
            true
          } else if (left.orderCnt == right.orderCnt) {
            left.payCnt > right.payCnt
          }
          else {
            false
          }
        }
        else {
          false
        }
      }
    ).take(10).foreach(println)


    sc.stop()
  }

  case class HotCategory(cid: String, var clickCnt: Int, var orderCnt: Int, var payCnt: Int)

  class HotCategoryACC extends AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] {
    private val hcMap = mutable.Map[String, HotCategory]()

    override def isZero: Boolean = {
      hcMap.isEmpty
    }

    override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] = {
      new HotCategoryACC
    }

    override def reset(): Unit = {
      hcMap.clear()
    }

    override def add(v: (String, String)): Unit = {
      val cid = v._1
      val actionType = v._2
      val category: HotCategory = hcMap.getOrElse(cid, HotCategory(cid, 0, 0, 0))
      if (actionType == "click") {
        category.clickCnt += 1
      } else if (actionType == "order") {
        category.orderCnt += 1
      } else if (actionType == "pay") {
        category.payCnt += 1
      }

      hcMap.update(cid, category)
    }

    override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]): Unit = {
      var map1 = this.hcMap
      var map2 = other.value

     /* map2.foreach {
        case (cid, hc) => {
          val category: HotCategory = map1.getOrElse(cid, HotCategory(cid, 0, 0, 0))
          category.clickCnt += hc.clickCnt
          category.orderCnt += hc.orderCnt
          category.payCnt += hc.payCnt
          map1.update(cid, category)
        }
      }*/

      map2.foldLeft(map1){
        (map,kv)=>{
          val cid: String = kv._1
          val hotCategory: HotCategory = kv._2
          val category: HotCategory = map.getOrElse(cid, HotCategory(cid, 0, 0, 0))
          category.clickCnt += hotCategory.clickCnt
          category.orderCnt += hotCategory.orderCnt
          category.payCnt += hotCategory.payCnt
          map.update(cid,category)
          map
        }
      }
    }

    override def value: mutable.Map[String, HotCategory] = {
      hcMap
    }
  }

}
